package thss.mro.bolt;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

import thss.mro.entity.Packet;
import thss.mro.entity.PacketField;
import thss.mro.entity.template.ProtocolTemplate;
import thss.mro.manager.ProtocolManager;
import thss.mro.manager.TemplateManager;
import thss.mro.util.Constants;
import thss.mro.util.TypeTransUtil;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

/*TODO: 解析正文的bolt
 * */
public class PacketBodyParse extends BaseRichBolt {
	OutputCollector _collector;
	
	@Override
	public void execute(Tuple tuple) {
		//从流中接收数据:bodyId-正文模板ID；bodyData-数据包正文部分
		String bodyTplId = tuple.getString(0);
		byte[] bodyData = (byte[])tuple.get("bodyData");
		//获得正文模板并解析正文
		ProtocolTemplate bodyTpl = TemplateManager.getTemplate(bodyTplId);
		Packet bodyPacket = ProtocolManager.parse(bodyTpl, bodyData);
		//根据正文模板类型确定流出方向
		if(bodyTplId.equals(Constants.TEMPLATE_ID_SCP_BODY)){
			//SCP包正文模板则解析出信元并发射信元流
			String headId = Constants.TEMPLATE_ID_SCPCELL_HEAD;
			List<Object> objLst = bodyPacket.getFieldValueLst("cell");
			System.out.println("SCP包含有"+objLst.size()+"个信元");
			for(int i=0;i<objLst.size();i++){
				byte[] cellData = (byte[])objLst.get(i);
				//回流：发送信元
				_collector.emit("cellStream",tuple, new Values(Constants.PROTOCOL_PORT_SCPCELL,cellData));
			}
		}else if(bodyTplId.startsWith(Constants.TEMPLATE_ID_SCPCELL_BODY)){
			System.out.println("信元");
			//是信元正文则传输数据流进行存储
			Map.Entry entry = null;
			Iterator it = bodyPacket.getFieldChain().entrySet().iterator();
			//循环为协议各字段生成随机数据
			while (it.hasNext()){ 
				entry = (Map.Entry)it.next();
				PacketField packFld = (PacketField)entry.getValue();
				String fieldName = packFld.getTemplateField().getName();
				Object valueObj = packFld.getValue();
				if(packFld.getTemplateField().getType()==Constants.PROTOCOL_FIELD_TYPE_CHAR){
					//字符串型
					String value = new String((byte[])valueObj);
					System.out.println("[CHAR]fieldName:"+fieldName+" ; value:"+value);
				}else if(packFld.getTemplateField().getType()==Constants.PROTOCOL_FIELD_TYPE_BIT){
					//布尔型
					System.out.println("[BOOL]fieldName:"+fieldName+" ; value:"+(Boolean)valueObj);
				}else if(packFld.getTemplateField().getType()==Constants.PROTOCOL_FIELD_TYPE_INT){
					//整型
					int value = TypeTransUtil.bytes2Int((byte[])valueObj);
					System.out.println("[INT]fieldName:"+fieldName+" ; value:"+value);
				}else{
					System.out.println("[BYTE]fieldName:"+fieldName+" ; value:"+TypeTransUtil.getBytesString((byte[])valueObj));
				}
			}
		}
		_collector.ack(tuple);
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		_collector = collector;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declareStream("cellStream",new Fields("port","data"));
		declarer.declareStream("dataStream",new Fields("data"));
	}

}
